package com.qf.work;

import com.qf.util.ConnUtil;
import com.rabbitmq.client.*;
import org.junit.Test;

import java.io.IOException;

public class Consumer {

    @Test
    public void consume1() throws Exception {
        //1. Conn
        Connection conn = ConnUtil.getConn();

        //2. Channel
        Channel channel = conn.createChannel();

        //3. Queue
        channel.queueDeclare(Publisher.QUEUE_NAME,true,false,false,null);

        //3.5 Qos
        channel.basicQos(1);

        //4. consume
        DefaultConsumer consume = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("消费者1号消费到消息:" + new String(body));
                //ack
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(Publisher.QUEUE_NAME,false,consume);
        //5. close
        System.in.read();
        channel.close();
        conn.close();
    }


    @Test
    public void consume2() throws Exception {
        //1. Conn
        Connection conn = ConnUtil.getConn();

        //2. Channel
        Channel channel = conn.createChannel();

        //3. Queue
        channel.queueDeclare(Publisher.QUEUE_NAME,true,false,false,null);

        //3.5 指定消费者的Qos
        channel.basicQos(1);

        //4. consume
        DefaultConsumer consume = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("消费者2号消费到消息:" + new String(body));
                // 手动ack
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(Publisher.QUEUE_NAME,false,consume);
        //5. close
        System.in.read();
        channel.close();
        conn.close();
    }
}
